/**
 * Copyright (c) 2025 myons Inc. All rights reserved.
 */
package cn.zmwh.manage.admin.modules.sync;

import cn.hutool.json.JSONUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 底层原生binlog监听
 *
 * @describe: 、
 * @author: dmzmwh 、
 * @time: 2025-04-02 14:56 、
 */
@Slf4j
public class BinaryLogClientInit {


    public static void main(String[] args) throws IOException, TimeoutException {
        BinaryLogClient client;
        /**
         * # 创建用户
         * CREATE USER binlog_user IDENTIFIED BY 'binlog#Replication2024!';
         *
         *
         * todo 很重要需要权限
         * GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'binlog_user'@'%';
         * FLUSH PRIVILEGES;
         */
        String hostname = "192.168.110.124";
        int port = 3306;
        String username = "zndy";
        String password = "zndy1234";
        // 创建 BinaryLogClient客户端
        client = new BinaryLogClient(hostname, port, username, password);
        // 这个 serviceId 不可重复
        client.setServerId(12);

        // 反序列化配置
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                // 将日期类型的数据反序列化成Long类型
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
        );

        client.setEventDeserializer(eventDeserializer);
        client.registerEventListener(event -> {
            EventType eventType = event.getHeader().getEventType();
            log.info("接收到事件类型: {}", JSONUtil.toJsonStr(eventType));
            log.info("接收到的完整事件: {}", JSONUtil.toJsonStr(event));
            log.info("======================================================================");
        });
        client.registerLifecycleListener(new BinaryLogClient.AbstractLifecycleListener() {
            @Override
            public void onConnect(BinaryLogClient client) {
                log.info("客户端连接到 mysql 服务器 client: {}", client);
            }

            @Override
            public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                ex.printStackTrace();
                log.info("客户端和 mysql 服务器 通讯失败 client: {}", client);
            }

            @Override
            public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                ex.printStackTrace();
                log.info("客户端序列化失败 client: {}", client);
            }

            @Override
            public void onDisconnect(BinaryLogClient client) {
                log.info("客户端断开 mysql 服务器链接 client: {}", client);
            }
        });
        // client.connect 在当前线程中进行解析binlog，会阻塞当前线程
        // client.connect(xxx) 会新开启一个线程，然后在这个线程中解析binlog
        client.connect(10000);
    }

}